package com.pk.flink.apps.consumerGroup.specifyPartition;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;


public class KafkaConsumerGroupConsumerApp {

    private static String SERVERS = "master:9092,slave1:9092,slave2:9092";
    private static String TOPIC = "pk-3-1";
    KafkaConsumer<String, String> consumer = null;

    public static void main(String[] args) {
        new KafkaConsumerGroupConsumerApp().run();
    }

    public void run(){
        Properties props = new Properties();
        props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer-group-id-" + System.currentTimeMillis());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        consumer = new KafkaConsumer<>(props);
//        consumer.subscribe(Arrays.asList(TOPIC));
        consumer.assign(Arrays.asList(new TopicPartition(TOPIC,0)));
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
                for (ConsumerRecord<String, String> record : records) {
                    String key = record.key();
                    String value = record.value();
                    String topic = record.topic();
                    Integer partition = record.partition();
                    long offset = record.offset();

                    System.out.println("key:" + key + ",value:" + value + ",topic:" + topic + ",partition:" + partition + ",offset:" + offset);
                }
            }
        }
        catch(Exception ex) {
            System.err.println(ex.getMessage());
        }
        finally {
            if (consumer != null) {
                consumer.close();
            }
        }
    }
}
